{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/aburkov/theLMbook/blob/main/emotion_classifier_CNN.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "yy0zjL_2ouOU",
        "outputId": "2bed3645-4617-4cf2-b5ec-4b998d55c869"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Using device: cuda\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "Downloading: 1.53GB [04:39, 5.89MB/s]                            \n",
            "Loading word vectors: 100%|██████████| 3000000/3000000 [00:55<00:00, 54420.35it/s]\n",
            "Epoch 1/2: 100%|██████████| 563/563 [00:07<00:00, 72.08it/s, Loss=0.759]\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Epoch [1/2], Test Accuracy: 0.8830\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "Epoch 2/2: 100%|██████████| 563/563 [00:06<00:00, 85.55it/s, Loss=0.248]\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Epoch [2/2], Test Accuracy: 0.9055\n"
          ]
        }
      ],
      "source": [
        "# Import required libraries\n",
        "import torch                             # PyTorch for tensor operations and deep learning\n",
        "import torch.nn as nn                    # Neural network module from PyTorch\n",
        "import numpy as np                       # NumPy for numerical operations\n",
        "import re                                # Regular expressions for text processing (if needed)\n",
        "import urllib.request                    # For downloading files from URLs\n",
        "import gzip                              # For handling compressed files\n",
        "import json                              # For parsing JSON data\n",
        "import requests                          # For making HTTP requests to download data\n",
        "import random                            # For shuffling data and setting random seeds\n",
        "import pickle                            # For saving and loading serialized objects\n",
        "import os                                # For file system operations\n",
        "from tqdm import tqdm                    # For displaying progress bars during loops\n",
        "from torch.utils.data import Dataset, DataLoader  # For creating custom datasets and data loaders in PyTorch\n",
        "\n",
        "\n",
        "def set_seed(seed):\n",
        "    \"\"\"\n",
        "    Sets random seeds for reproducibility across different libraries.\n",
        "\n",
        "    Args:\n",
        "        seed (int): Seed value for random number generation.\n",
        "    \"\"\"\n",
        "    random.seed(seed)                      # Set seed for Python's built-in random module\n",
        "    torch.manual_seed(seed)                # Set seed for CPU operations in PyTorch\n",
        "    torch.cuda.manual_seed_all(seed)       # Set seed for all GPUs\n",
        "    torch.backends.cudnn.deterministic = True  # Use deterministic algorithms for cuDNN\n",
        "    torch.backends.cudnn.benchmark = False     # Disable cuDNN auto-tuner for consistent behavior\n",
        "\n",
        "\n",
        "class Tokenizer:\n",
        "    \"\"\"\n",
        "    Basic tokenizer that splits text on whitespace.\n",
        "    \"\"\"\n",
        "    def tokenize(self, text):\n",
        "        \"\"\"\n",
        "        Splits the input text into tokens based on whitespace.\n",
        "\n",
        "        Args:\n",
        "            text (str): Input text string to tokenize.\n",
        "\n",
        "        Returns:\n",
        "            list: List of word tokens.\n",
        "        \"\"\"\n",
        "        words = text.split()             # Split text by whitespace\n",
        "        return words\n",
        "\n",
        "\n",
        "class Embedder:\n",
        "    \"\"\"\n",
        "    Embedder that converts tokens into their corresponding embeddings.\n",
        "\n",
        "    Attributes:\n",
        "        embeddings (dict): Dictionary mapping words to their vector representations.\n",
        "        emb_dim (int): Dimensionality of the embeddings.\n",
        "        seq_len (int): Fixed sequence length for input text.\n",
        "    \"\"\"\n",
        "    def __init__(self, embeddings, emb_dim, seq_len):\n",
        "        \"\"\"\n",
        "        Initializes the Embedder with embeddings, embedding dimension, and sequence length.\n",
        "\n",
        "        Args:\n",
        "            embeddings (dict): Pre-loaded word embeddings.\n",
        "            emb_dim (int): Dimension of the embeddings.\n",
        "            seq_len (int): Maximum number of tokens to consider.\n",
        "        \"\"\"\n",
        "        self.embeddings = embeddings\n",
        "        self.emb_dim = emb_dim\n",
        "        self.seq_len = seq_len\n",
        "\n",
        "    def embed(self, tokens):\n",
        "        \"\"\"\n",
        "        Converts a list of tokens into a tensor of embeddings.\n",
        "\n",
        "        Tokens are looked up in the embeddings dictionary. If a token is not found,\n",
        "        a zero vector is used. The sequence is truncated or padded to match the fixed sequence length.\n",
        "\n",
        "        Args:\n",
        "            tokens (list): List of token strings.\n",
        "\n",
        "        Returns:\n",
        "            torch.Tensor: Tensor of shape (seq_len, emb_dim) representing the embedded tokens.\n",
        "        \"\"\"\n",
        "        embeddings = []\n",
        "        # Process each token up to the maximum sequence length\n",
        "        for word in tokens[:self.seq_len]:\n",
        "            if word in self.embeddings:\n",
        "                embeddings.append(torch.tensor(self.embeddings[word]))\n",
        "            elif word.lower() in self.embeddings:\n",
        "                embeddings.append(torch.tensor(self.embeddings[word.lower()]))\n",
        "            else:\n",
        "                # Use a zero vector for words not found in the embeddings\n",
        "                embeddings.append(torch.zeros(self.emb_dim))\n",
        "\n",
        "        # Pad sequence with zero vectors if the number of tokens is less than seq_len\n",
        "        if len(embeddings) < self.seq_len:\n",
        "            padding_size = self.seq_len - len(embeddings)\n",
        "            embeddings.extend([torch.zeros(self.emb_dim)] * padding_size)\n",
        "\n",
        "        # Stack list of tensors into a single tensor of shape (seq_len, emb_dim)\n",
        "        return torch.stack(embeddings)\n",
        "\n",
        "\n",
        "def load_embeddings(url, filename=\"vectors.dat\"):\n",
        "    \"\"\"\n",
        "    Downloads and loads word embeddings from a gzipped file.\n",
        "\n",
        "    If the file does not exist locally, it is downloaded from the provided URL.\n",
        "    The file is expected to have a header indicating vocabulary size and embedding dimension,\n",
        "    followed by each word and its corresponding binary vector.\n",
        "\n",
        "    Args:\n",
        "        url (str): URL to download the embeddings from.\n",
        "        filename (str): Local filename to save/load the embeddings.\n",
        "\n",
        "    Returns:\n",
        "        tuple: A tuple containing:\n",
        "            - vectors (dict): Mapping from words to their embedding vectors (as NumPy arrays).\n",
        "            - emb_dim (int): Dimensionality of the embedding vectors.\n",
        "    \"\"\"\n",
        "    # Check if the embeddings file exists locally\n",
        "    if not os.path.exists(filename):\n",
        "        # Download the embeddings file with a progress bar\n",
        "        with tqdm(unit=\"B\", unit_scale=True, unit_divisor=1024, desc=\"Downloading\") as progress_bar:\n",
        "            def report_hook(count, block_size, total_size):\n",
        "                if total_size != -1:\n",
        "                    progress_bar.total = total_size\n",
        "                progress_bar.update(block_size)\n",
        "            urllib.request.urlretrieve(url, filename, reporthook=report_hook)\n",
        "    else:\n",
        "        print(f\"File {filename} already exists. Skipping download.\")\n",
        "\n",
        "    # Open the gzipped embeddings file in binary read mode\n",
        "    with gzip.open(filename, \"rb\") as f:\n",
        "        # Read header line to get vocabulary size and embedding dimension\n",
        "        header = f.readline()\n",
        "        vocab_size, emb_dim = map(int, header.split())\n",
        "\n",
        "        vectors = {}\n",
        "        # Calculate the number of bytes for each embedding vector\n",
        "        binary_len = np.dtype(\"float32\").itemsize * emb_dim\n",
        "\n",
        "        # Read each word and its corresponding embedding vector with a progress bar\n",
        "        with tqdm(total=vocab_size, desc=\"Loading word vectors\") as pbar:\n",
        "            for _ in range(vocab_size):\n",
        "                word = []\n",
        "                # Read characters one by one until a space is encountered (indicating end of word)\n",
        "                while True:\n",
        "                    ch = f.read(1)\n",
        "                    if ch == b\" \":\n",
        "                        word = b\"\".join(word).decode(\"utf-8\")\n",
        "                        break\n",
        "                    if ch != b\"\\n\":\n",
        "                        word.append(ch)\n",
        "\n",
        "                # Read the binary vector data and convert it into a NumPy array of type float32\n",
        "                vector = np.frombuffer(f.read(binary_len), dtype=\"float32\")\n",
        "                vectors[word] = vector\n",
        "                pbar.update(1)\n",
        "\n",
        "    return vectors, emb_dim\n",
        "\n",
        "\n",
        "def load_and_split_data(url, test_ratio=0.1):\n",
        "    \"\"\"\n",
        "    Downloads, decompresses, and splits the dataset into training and testing sets.\n",
        "\n",
        "    The dataset is expected to be a gzipped file where each line is a JSON object.\n",
        "\n",
        "    Args:\n",
        "        url (str): URL to download the dataset from.\n",
        "        test_ratio (float): Proportion of data to be used as the test set.\n",
        "\n",
        "    Returns:\n",
        "        tuple: A tuple containing:\n",
        "            - train_data (list): List of training examples.\n",
        "            - test_data (list): List of testing examples.\n",
        "    \"\"\"\n",
        "    # Download the dataset from the provided URL\n",
        "    response = requests.get(url)\n",
        "    # Decompress the gzipped content and decode it to a string\n",
        "    content = gzip.decompress(response.content).decode()\n",
        "    # Parse each line as a JSON object\n",
        "    data = [json.loads(line) for line in content.splitlines()]\n",
        "    # Shuffle the data to ensure a random distribution\n",
        "    random.shuffle(data)\n",
        "    # Determine the split index based on the test_ratio\n",
        "    split_index = int(len(data) * (1 - test_ratio))\n",
        "    return data[:split_index], data[split_index:]\n",
        "\n",
        "\n",
        "def download_and_prepare_data(data_url, vectors_url, seq_len, batch_size):\n",
        "    \"\"\"\n",
        "    Downloads and prepares the dataset and word embeddings for training and evaluation.\n",
        "\n",
        "    This function downloads the text dataset and word embeddings, creates label mappings,\n",
        "    initializes the tokenizer and embedder, and returns data loaders for training and testing.\n",
        "\n",
        "    Args:\n",
        "        data_url (str): URL to download the text dataset.\n",
        "        vectors_url (str): URL to download the word embeddings.\n",
        "        seq_len (int): Fixed sequence length for token embeddings.\n",
        "        batch_size (int): Batch size for the data loaders.\n",
        "\n",
        "    Returns:\n",
        "        tuple: A tuple containing:\n",
        "            - train_loader (DataLoader): DataLoader for training data.\n",
        "            - test_loader (DataLoader): DataLoader for testing data.\n",
        "            - id_to_label (dict): Mapping from label IDs to label names.\n",
        "            - num_classes (int): Number of unique classes.\n",
        "            - emb_dim (int): Dimensionality of the word embeddings.\n",
        "    \"\"\"\n",
        "    # Load and split the dataset into training and testing splits\n",
        "    train_split, test_split = load_and_split_data(data_url, test_ratio=0.1)\n",
        "\n",
        "    # Load pre-trained word embeddings and get the embedding dimension\n",
        "    embeddings, emb_dim = load_embeddings(vectors_url)\n",
        "\n",
        "    # Create mappings between labels and their numeric IDs using the training data\n",
        "    label_to_id, id_to_label, num_classes = create_label_mappings(train_split)\n",
        "\n",
        "    # Initialize the tokenizer and embedder with the loaded embeddings and sequence length\n",
        "    tokenizer = Tokenizer()\n",
        "    embedder = Embedder(embeddings, emb_dim, seq_len)\n",
        "\n",
        "    # Create DataLoaders for both training and testing datasets\n",
        "    train_loader, test_loader = create_data_loaders(\n",
        "        train_split, test_split,\n",
        "        tokenizer, embedder,\n",
        "        label_to_id, batch_size\n",
        "    )\n",
        "\n",
        "    return (train_loader, test_loader, id_to_label, num_classes, emb_dim)\n",
        "\n",
        "\n",
        "class TextClassificationDataset(Dataset):\n",
        "    \"\"\"\n",
        "    PyTorch Dataset for text classification.\n",
        "\n",
        "    This dataset converts raw text and label data into a format that can be fed into the model.\n",
        "    It tokenizes text and then embeds it using the provided tokenizer and embedder.\n",
        "\n",
        "    Args:\n",
        "        data (list): List of dictionaries containing \"text\" and \"label\" keys.\n",
        "        tokenizer (Tokenizer): Tokenizer instance to split text into tokens.\n",
        "        embedder (Embedder): Embedder instance to convert tokens into embeddings.\n",
        "        label_to_id (dict): Mapping from label strings to numeric IDs.\n",
        "    \"\"\"\n",
        "    def __init__(self, data, tokenizer, embedder, label_to_id):\n",
        "        # Extract texts and convert labels to their corresponding IDs\n",
        "        self.texts = [item[\"text\"] for item in data]\n",
        "        self.label_ids = [label_to_id[item[\"label\"]] for item in data]\n",
        "        self.tokenizer = tokenizer\n",
        "        self.embedder = embedder\n",
        "\n",
        "    def __len__(self):\n",
        "        \"\"\"\n",
        "        Returns the total number of examples in the dataset.\n",
        "        \"\"\"\n",
        "        return len(self.texts)\n",
        "\n",
        "    def __getitem__(self, idx):\n",
        "        \"\"\"\n",
        "        Retrieves the embedded text and label for the example at the specified index.\n",
        "\n",
        "        Args:\n",
        "            idx (int): Index of the example to retrieve.\n",
        "\n",
        "        Returns:\n",
        "            tuple: A tuple containing:\n",
        "                - embeddings (torch.Tensor): Tensor of shape (seq_len, emb_dim).\n",
        "                - label (torch.Tensor): Tensor containing the label ID.\n",
        "        \"\"\"\n",
        "        # Tokenize the text at the given index\n",
        "        tokens = self.tokenizer.tokenize(self.texts[idx])\n",
        "        # Convert tokens into embeddings using the embedder\n",
        "        embeddings = self.embedder.embed(tokens)\n",
        "        # Return the embeddings and corresponding label as a tensor\n",
        "        return embeddings, torch.tensor(self.label_ids[idx], dtype=torch.long)\n",
        "\n",
        "\n",
        "class CNNTextClassifier(nn.Module):\n",
        "    \"\"\"\n",
        "    Convolutional Neural Network for text classification.\n",
        "\n",
        "    This model applies two 1D convolutional layers followed by fully connected layers\n",
        "    to classify input text based on its embedded representation.\n",
        "\n",
        "    Args:\n",
        "        emb_dim (int): Dimensionality of the input embeddings.\n",
        "        num_classes (int): Number of target classes for classification.\n",
        "        seq_len (int): Fixed sequence length of the input text.\n",
        "        id_to_label (dict): Mapping from label IDs to label names.\n",
        "    \"\"\"\n",
        "    def __init__(self, emb_dim, num_classes, seq_len, id_to_label):\n",
        "        super().__init__()\n",
        "        # Save model configuration for later use (e.g., during inference)\n",
        "        self.config = {\n",
        "            \"emb_dim\": emb_dim,\n",
        "            \"num_classes\": num_classes,\n",
        "            \"seq_len\": seq_len,\n",
        "            \"id_to_label\": id_to_label\n",
        "        }\n",
        "        # First convolutional layer: input channels = emb_dim, output channels = 512\n",
        "        self.conv1 = nn.Conv1d(emb_dim, 512, kernel_size=3, padding=1)\n",
        "        # Second convolutional layer: input channels = 512, output channels = 256\n",
        "        self.conv2 = nn.Conv1d(512, 256, kernel_size=3, padding=1)\n",
        "        # Fully connected layer to reduce flattened features to 128 units\n",
        "        self.fc1 = nn.Linear(256 * seq_len, 128)\n",
        "        # Output layer mapping to the number of classes\n",
        "        self.fc2 = nn.Linear(128, num_classes)\n",
        "        # ReLU activation function\n",
        "        self.relu = nn.ReLU()\n",
        "\n",
        "    def forward(self, x):\n",
        "        \"\"\"\n",
        "        Defines the forward pass of the CNN model.\n",
        "\n",
        "        Args:\n",
        "            x (torch.Tensor): Input tensor of shape (batch_size, seq_len, emb_dim).\n",
        "\n",
        "        Returns:\n",
        "            torch.Tensor: Output logits tensor of shape (batch_size, num_classes).\n",
        "        \"\"\"\n",
        "        x = x.permute(0, 2, 1)           # Rearrange dimensions to (batch_size, emb_dim, seq_len)\n",
        "        x = self.relu(self.conv1(x))     # Apply first convolution and ReLU activation\n",
        "        x = self.relu(self.conv2(x))     # Apply second convolution and ReLU activation\n",
        "        x = x.flatten(start_dim=1)       # Flatten features for the fully connected layers\n",
        "        x = self.fc1(x)                  # Apply first fully connected layer\n",
        "        return self.fc2(x)               # Return output logits from the final layer\n",
        "\n",
        "\n",
        "def calculate_accuracy(model, dataloader, device):\n",
        "    \"\"\"\n",
        "    Evaluates the model's accuracy on the provided dataset.\n",
        "\n",
        "    Args:\n",
        "        model (nn.Module): Trained model.\n",
        "        dataloader (DataLoader): DataLoader for the dataset to evaluate.\n",
        "        device: Device on which computations are performed.\n",
        "\n",
        "    Returns:\n",
        "        float: Accuracy as a fraction of correct predictions.\n",
        "    \"\"\"\n",
        "    model.eval()                       # Set model to evaluation mode\n",
        "    correct = 0\n",
        "    total = 0\n",
        "    with torch.no_grad():              # Disable gradient calculations for efficiency\n",
        "        for batch in dataloader:\n",
        "            embeddings, labels = batch\n",
        "            embeddings = embeddings.to(device)\n",
        "            labels = labels.to(device)\n",
        "            outputs = model(embeddings)           # Forward pass\n",
        "            _, predicted = torch.max(outputs, 1)   # Get predicted class indices\n",
        "            total += labels.size(0)\n",
        "            correct += (predicted == labels).sum().item()\n",
        "    accuracy = correct / total\n",
        "    model.train()                      # Set model back to training mode\n",
        "    return accuracy\n",
        "\n",
        "\n",
        "def create_label_mappings(train_dataset):\n",
        "    \"\"\"\n",
        "    Creates mappings between label strings and numeric IDs based on the training dataset.\n",
        "\n",
        "    Args:\n",
        "        train_dataset (list): List of training examples with a \"label\" field.\n",
        "\n",
        "    Returns:\n",
        "        tuple: A tuple containing:\n",
        "            - label_to_id (dict): Mapping from label string to numeric ID.\n",
        "            - id_to_label (dict): Mapping from numeric ID to label string.\n",
        "            - num_classes (int): Total number of unique classes.\n",
        "    \"\"\"\n",
        "    # Extract and sort unique labels from the training data\n",
        "    unique_labels = sorted(set(item[\"label\"] for item in train_dataset))\n",
        "    # Create a mapping from each label to a unique integer ID\n",
        "    label_to_id = {label: i for i, label in enumerate(unique_labels)}\n",
        "    # Create the reverse mapping from ID to label\n",
        "    id_to_label = {i: label for label, i in label_to_id.items()}\n",
        "    return label_to_id, id_to_label, len(unique_labels)\n",
        "\n",
        "\n",
        "def create_data_loaders(train_split, test_split, tokenizer, embedder, label_to_id, batch_size):\n",
        "    \"\"\"\n",
        "    Creates PyTorch DataLoaders for training and testing datasets.\n",
        "\n",
        "    Args:\n",
        "        train_split (list): List of training examples.\n",
        "        test_split (list): List of testing examples.\n",
        "        tokenizer (Tokenizer): Tokenizer instance for processing text.\n",
        "        embedder (Embedder): Embedder instance for converting tokens to embeddings.\n",
        "        label_to_id (dict): Mapping from label strings to numeric IDs.\n",
        "        batch_size (int): Batch size for the DataLoaders.\n",
        "\n",
        "    Returns:\n",
        "        tuple: A tuple containing:\n",
        "            - train_loader (DataLoader): DataLoader for the training dataset.\n",
        "            - test_loader (DataLoader): DataLoader for the testing dataset.\n",
        "    \"\"\"\n",
        "    # Initialize the custom dataset for training and testing data\n",
        "    train_dataset = TextClassificationDataset(train_split, tokenizer, embedder, label_to_id)\n",
        "    test_dataset = TextClassificationDataset(test_split, tokenizer, embedder, label_to_id)\n",
        "    # Create DataLoaders; enable shuffling for training data\n",
        "    train_loader = DataLoader(train_dataset, batch_size, shuffle=True)\n",
        "    test_loader = DataLoader(test_dataset, batch_size)\n",
        "    return train_loader, test_loader\n",
        "\n",
        "\n",
        "def save_model(model, prefix):\n",
        "    \"\"\"\n",
        "    Saves the trained model's state dictionary and configuration to disk.\n",
        "\n",
        "    Args:\n",
        "        model (nn.Module): Trained model to be saved.\n",
        "        prefix (str): Prefix for the saved file name.\n",
        "    \"\"\"\n",
        "    # Save the model's state and configuration as a checkpoint file\n",
        "    torch.save({\n",
        "        \"state_dict\": model.state_dict(),\n",
        "        \"config\": model.config\n",
        "    }, f\"{prefix}_model.pth\")\n",
        "\n",
        "\n",
        "def load_model(prefix):\n",
        "    \"\"\"\n",
        "    Loads a saved model from disk and prepares it for evaluation.\n",
        "\n",
        "    Args:\n",
        "        prefix (str): Prefix used during model saving.\n",
        "\n",
        "    Returns:\n",
        "        nn.Module: The loaded CNNTextClassifier model in evaluation mode.\n",
        "    \"\"\"\n",
        "    # Load checkpoint containing model state and configuration\n",
        "    checkpoint = torch.load(f\"{prefix}_model.pth\", map_location=torch.device(\"cpu\"))\n",
        "    config = checkpoint[\"config\"]\n",
        "    # Reinitialize the model using the saved configuration\n",
        "    model = CNNTextClassifier(\n",
        "        emb_dim=config[\"emb_dim\"],\n",
        "        num_classes=config[\"num_classes\"],\n",
        "        seq_len=config[\"seq_len\"],\n",
        "        id_to_label=config[\"id_to_label\"]\n",
        "    )\n",
        "    # Load the saved weights into the model\n",
        "    model.load_state_dict(checkpoint[\"state_dict\"])\n",
        "    model.eval()  # Set the model to evaluation mode\n",
        "\n",
        "    return model\n",
        "\n",
        "\n",
        "def test_model(model, test_input, tokenizer=None, embedder=None):\n",
        "    \"\"\"\n",
        "    Tests the model on a single input text and prints the predicted label.\n",
        "\n",
        "    Args:\n",
        "        model (nn.Module): Trained text classification model.\n",
        "        test_input (str): Input text to classify.\n",
        "        tokenizer (Tokenizer, optional): Tokenizer instance. If None, a new one is created.\n",
        "        embedder (Embedder, optional): Embedder instance. If None, embeddings are loaded and a new embedder is created.\n",
        "\n",
        "    Notes:\n",
        "        This function prints the input text along with the predicted emotion.\n",
        "    \"\"\"\n",
        "    # Initialize tokenizer if not provided\n",
        "    if not tokenizer:\n",
        "        tokenizer = Tokenizer()\n",
        "    # Initialize embedder if not provided by loading embeddings using global vectors_url and seq_len\n",
        "    if not embedder:\n",
        "        embeddings, emb_dim = load_embeddings(vectors_url)\n",
        "        embedder = Embedder(embeddings, emb_dim, seq_len)\n",
        "\n",
        "    # Determine the device from the model's parameters\n",
        "    device = next(model.parameters()).device\n",
        "    model.eval()  # Set model to evaluation mode\n",
        "    with torch.no_grad():\n",
        "        # Tokenize the input text\n",
        "        tokens = tokenizer.tokenize(test_input)\n",
        "        # Convert tokens into embeddings\n",
        "        embeddings = embedder.embed(tokens)\n",
        "        # Convert embeddings to a float tensor, add batch dimension, and move to the correct device\n",
        "        embeddings = torch.tensor(embeddings, dtype=torch.float32).unsqueeze(0).to(device)\n",
        "\n",
        "        # Perform a forward pass through the model to get predictions\n",
        "        outputs = model(embeddings)\n",
        "        _, predicted = torch.max(outputs.data, 1)\n",
        "\n",
        "        # Map the predicted numeric label to the actual label string\n",
        "        predicted_label = model.config[\"id_to_label\"][predicted.item()]\n",
        "\n",
        "    print(f\"Input: {test_input}\")\n",
        "    print(f\"Predicted emotion: {predicted_label}\")\n",
        "\n",
        "\n",
        "def set_hyperparameters():\n",
        "    \"\"\"\n",
        "    Defines and returns hyperparameters for training.\n",
        "\n",
        "    Returns:\n",
        "        tuple: A tuple containing:\n",
        "            - num_epochs (int): Number of training epochs.\n",
        "            - seq_len (int): Fixed sequence length for input text.\n",
        "            - batch_size (int): Batch size for training.\n",
        "            - learning_rate (float): Learning rate for the optimizer.\n",
        "    \"\"\"\n",
        "    num_epochs = 2\n",
        "    seq_len = 100\n",
        "    batch_size = 32\n",
        "    learning_rate = 0.001\n",
        "    return num_epochs, seq_len, batch_size, learning_rate\n",
        "\n",
        "\n",
        "if __name__ == \"__main__\":\n",
        "    # Set random seeds for reproducibility\n",
        "    set_seed(42)\n",
        "    # Determine computation device: use GPU if available, otherwise CPU\n",
        "    device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
        "    print(f\"Using device: {device}\")\n",
        "\n",
        "    # URLs for downloading the dataset and word embeddings\n",
        "    data_url = \"https://www.thelmbook.com/data/emotions\"\n",
        "    vectors_url = \"https://www.thelmbook.com/data/word-vectors\"\n",
        "\n",
        "    # Set training hyperparameters\n",
        "    num_epochs, seq_len, batch_size, learning_rate = set_hyperparameters()\n",
        "\n",
        "    # Download and prepare data loaders, label mappings, and embedding dimensions\n",
        "    train_loader, test_loader, id_to_label, num_classes, emb_dim = \\\n",
        "        download_and_prepare_data(data_url, vectors_url, seq_len, batch_size)\n",
        "\n",
        "    # Initialize the CNN text classifier model with the embedding dimension and label mappings\n",
        "    model = CNNTextClassifier(emb_dim, num_classes, seq_len, id_to_label)\n",
        "    model = model.to(device)  # Move model to the appropriate device\n",
        "\n",
        "    # Define the loss function and optimizer\n",
        "    criterion = nn.CrossEntropyLoss()                # Cross-entropy loss for classification\n",
        "    optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)  # AdamW optimizer\n",
        "\n",
        "    # Training loop over multiple epochs\n",
        "    for epoch in range(num_epochs):\n",
        "        model.train()                  # Set model to training mode\n",
        "        total_loss = 0\n",
        "        num_batches = 0\n",
        "        # Initialize progress bar for the current epoch\n",
        "        progress_bar = tqdm(train_loader, desc=f\"Epoch {epoch+1}/{num_epochs}\")\n",
        "        for batch in progress_bar:\n",
        "            batch_embeddings, batch_labels = batch\n",
        "            batch_embeddings = batch_embeddings.to(device)\n",
        "            batch_labels = batch_labels.to(device)\n",
        "            optimizer.zero_grad()      # Reset gradients before backpropagation\n",
        "            outputs = model(batch_embeddings)   # Forward pass through the model\n",
        "            loss = criterion(outputs, batch_labels)  # Compute loss\n",
        "            loss.backward()            # Backpropagation\n",
        "            optimizer.step()           # Update model parameters\n",
        "\n",
        "            total_loss += loss.item()\n",
        "            num_batches += 1\n",
        "\n",
        "            # Update progress bar with the current average loss\n",
        "            progress_bar.set_postfix({\"Loss\": total_loss / num_batches})\n",
        "\n",
        "        avg_loss = total_loss / num_batches\n",
        "\n",
        "        # Evaluate model accuracy on the test set after each epoch\n",
        "        test_acc = calculate_accuracy(model, test_loader, device)\n",
        "        print(f\"Epoch [{epoch+1}/{num_epochs}], Test Accuracy: {test_acc:.4f}\")\n",
        "\n",
        "    # Save the trained model to disk with the specified prefix\n",
        "    model_name = \"CNN_classifier\"\n",
        "    save_model(model, model_name)"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "if __name__ == \"__main__\":\n",
        "    # Load the previously saved model using the specified model name prefix.\n",
        "    # The 'load_model' function reads the checkpoint and reconstructs the CNNTextClassifier.\n",
        "    loaded_model = load_model(model_name)\n",
        "\n",
        "    # Load the word embeddings from the provided URL.\n",
        "    # This returns a dictionary mapping words to their embedding vectors and the embedding dimension.\n",
        "    embeddings, emb_dim = load_embeddings(vectors_url)\n",
        "\n",
        "    # Initialize the tokenizer.\n",
        "    # The Tokenizer here simply splits text into tokens based on whitespace.\n",
        "    tokenizer = Tokenizer()\n",
        "\n",
        "    # Create an embedder instance using the loaded embeddings, embedding dimension, and fixed sequence length.\n",
        "    # The Embedder will convert tokenized text into a tensor of embeddings suitable for the model.\n",
        "    embedder = Embedder(embeddings, emb_dim, seq_len)\n",
        "\n",
        "    # Define a sample input text to classify.\n",
        "    test_input = \"I'm so happy to be able to train a text classifier!\"\n",
        "\n",
        "    # Use the test_model function to evaluate the loaded model on the sample input.\n",
        "    # This function tokenizes the text, converts it to embeddings, performs a forward pass through the model,\n",
        "    # and prints out the predicted label.\n",
        "    test_model(loaded_model, test_input, tokenizer, embedder)\n"
      ],
      "metadata": {
        "id": "CXhorMSZSkdZ",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "07d4a939-4b65-4fd9-c0ec-65839db14e24"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "<ipython-input-1-441c3652facc>:436: FutureWarning: You are using `torch.load` with `weights_only=False` (the current default value), which uses the default pickle module implicitly. It is possible to construct malicious pickle data which will execute arbitrary code during unpickling (See https://github.com/pytorch/pytorch/blob/main/SECURITY.md#untrusted-models for more details). In a future release, the default value for `weights_only` will be flipped to `True`. This limits the functions that could be executed during unpickling. Arbitrary objects will no longer be allowed to be loaded via this mode unless they are explicitly allowlisted by the user via `torch.serialization.add_safe_globals`. We recommend you start setting `weights_only=True` for any use case where you don't have full control of the loaded file. Please open an issue on GitHub for any issues related to this experimental feature.\n",
            "  checkpoint = torch.load(f\"{prefix}_model.pth\", map_location=torch.device(\"cpu\"))\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "File vectors.dat already exists. Skipping download.\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "Loading word vectors: 100%|██████████| 3000000/3000000 [00:54<00:00, 54728.27it/s]\n",
            "<ipython-input-1-441c3652facc>:482: UserWarning: To copy construct from a tensor, it is recommended to use sourceTensor.clone().detach() or sourceTensor.clone().detach().requires_grad_(True), rather than torch.tensor(sourceTensor).\n",
            "  embeddings = torch.tensor(embeddings, dtype=torch.float32).unsqueeze(0).to(device)\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Input: I'm so happy to be able to train a text classifier!\n",
            "Predicted emotion: joy\n"
          ]
        }
      ]
    }
  ],
  "metadata": {
    "accelerator": "GPU",
    "colab": {
      "gpuType": "A100",
      "provenance": [],
      "authorship_tag": "ABX9TyMKgD05suI0YlOL6D9FluVO",
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}